package sqlengine.queryExecution;

import static org.junit.Assert.*;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

import org.junit.Test;

import sqlengine.conditions.JoinCondition;
import sqlengine.conditions.NonJoinCondition;
import sqlengine.database.DataFile;
import sqlengine.database.DataManager;
import sqlengine.database.impl.Record;

public class BasicOperationTest {

	@Test
	public void testTableScan() {
		String fileName = "A";
		Map<String, Integer> descriptor = new HashMap<String, Integer>();
		descriptor.put("age", 50);
		descriptor.put("height", 50);
		DataFile dataFile = DataManager.createFile(fileName, descriptor);
		Map<String, String> record = new HashMap<String, String>();
		record.put("age", "16");
		record.put("height", "170");
		dataFile.insertRecord(record);
		record = new HashMap<String, String>();
		record.put("age", "18");
		record.put("height", "172");
		dataFile.insertRecord(record);
		ArrayList<Record> result = BasicOperation.tableScan("A");
		assertEquals("size 2", result.size(), 2);
		assertEquals("content", result.get(0).getKey("age"), "16");
		assertEquals("content", result.get(0).getKey("height"), "170");
		assertEquals("content", result.get(1).getKey("age"), "18");
		assertEquals("content", result.get(1).getKey("height"), "172");
		DataManager.exit();
	}

	@Test
	public void testIndexScan() {
		try {
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			dataFile.createIndex("_A.age", "age");
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "170");
			dataFile.insertRecord(record);

			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "172");
			dataFile.insertRecord(record);

			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "174");
			dataFile.insertRecord(record);

			ArrayList<Record> result;
			result = BasicOperation.indexScan(new NonJoinCondition("A", "age",
					"18"));

			assertEquals("size 2", result.size(), 2);
			assertEquals("content", result.get(0).getKey("age"), "18");
			assertEquals("content", result.get(0).getKey("height"), "172");
			assertEquals("content", result.get(1).getKey("age"), "18");
			assertEquals("content", result.get(1).getKey("height"), "174");

			result = BasicOperation.indexScan(new NonJoinCondition("A", "age",
					"16"));
			assertEquals("size 2", result.size(), 1);
			assertEquals("content", result.get(0).getKey("age"), "16");
			assertEquals("content", result.get(0).getKey("height"), "170");

			result = BasicOperation.indexScan(new NonJoinCondition("A", "age",
					"14"));
			assertEquals("size 2", result.size(), 0);
			DataManager.exit();
		} catch (BasicOperationException e) {
			// TODO Auto-generated catch block
			fail("should not throw exception");
		}
	}

	@Test
	public void testCartesian_product() {
		ArrayList<Record> records1 = new ArrayList<Record>();
		Record tmpRecord;
		Map<String, String> record;
		Map<String, Integer> schema = new HashMap<String, Integer>();
		schema.put("age", 50);
		schema.put("height", 50);

		record = new HashMap<String, String>();
		record.put("age", "16");
		record.put("height", "170");
		tmpRecord = new Record();
		tmpRecord.setRecord(record, schema);
		records1.add(tmpRecord);

		record = new HashMap<String, String>();
		record.put("age", "18");
		record.put("height", "172");
		tmpRecord = new Record();
		tmpRecord.setRecord(record, schema);
		records1.add(tmpRecord);

		ArrayList<Record> records2 = new ArrayList<Record>();
		schema = new HashMap<String, Integer>();
		schema.put("sex", 50);
		schema.put("weight", 50);

		record = new HashMap<String, String>();
		record.put("sex", "male");
		record.put("weight", "70");
		tmpRecord = new Record();
		tmpRecord.setRecord(record, schema);
		records2.add(tmpRecord);
		record = new HashMap<String, String>();
		record.put("sex", "female");
		record.put("weight", "75");
		tmpRecord = new Record();
		tmpRecord.setRecord(record, schema);
		records2.add(tmpRecord);

		ArrayList<Record> result = BasicOperation.cartesian_product(records1,
				records2);
		assertEquals("size 8", result.size(), 8);

		assertEquals("content", result.get(0).getKey("age"), "16");
		assertEquals("content", result.get(0).getKey("height"), "170");
		assertEquals("content", result.get(1).getKey("sex"), "male");
		assertEquals("content", result.get(1).getKey("weight"), "70");

		assertEquals("content", result.get(2).getKey("age"), "16");
		assertEquals("content", result.get(2).getKey("height"), "170");
		assertEquals("content", result.get(3).getKey("sex"), "female");
		assertEquals("content", result.get(3).getKey("weight"), "75");

		assertEquals("content", result.get(4).getKey("age"), "18");
		assertEquals("content", result.get(4).getKey("height"), "172");
		assertEquals("content", result.get(5).getKey("sex"), "male");
		assertEquals("content", result.get(5).getKey("weight"), "70");

		assertEquals("content", result.get(6).getKey("age"), "18");
		assertEquals("content", result.get(6).getKey("height"), "172");
		assertEquals("content", result.get(7).getKey("sex"), "female");
		assertEquals("content", result.get(7).getKey("weight"), "75");

		DataManager.exit();
	}

	@Test
	public void testNested_loops_join() {
		String fileName = "A";
		Map<String, Integer> descriptor = new HashMap<String, Integer>();
		descriptor.put("age", 50);
		descriptor.put("height", 50);
		DataFile dataFile = DataManager.createFile(fileName, descriptor);
		Map<String, String> record = new HashMap<String, String>();
		record.put("age", "170");
		record.put("height", "170");
		dataFile.insertRecord(record);
		record = new HashMap<String, String>();
		record.put("age", "18");
		record.put("height", "18");
		dataFile.insertRecord(record);
		record = new HashMap<String, String>();
		record.put("age", "16");
		record.put("height", "172");
		dataFile.insertRecord(record);
		record = new HashMap<String, String>();
		record.put("age", "18");
		record.put("height", "170");
		dataFile.insertRecord(record);

		fileName = "B";
		descriptor = new HashMap<String, Integer>();
		descriptor.put("sex", 50);
		descriptor.put("old", 50);
		descriptor.put("year", 50);
		dataFile = DataManager.createFile(fileName, descriptor);
		record = new HashMap<String, String>();
		record.put("sex", "male");
		record.put("old", "18");
		record.put("year", "18");
		dataFile.insertRecord(record);
		record = new HashMap<String, String>();
		record.put("sex", "female");
		record.put("old", "16");
		record.put("year", "16");
		dataFile.insertRecord(record);
		record = new HashMap<String, String>();
		record.put("sex", "male");
		record.put("old", "16");
		record.put("year", "18");
		dataFile.insertRecord(record);

		ArrayList<Record> result = BasicOperation.nested_loops_join(
				BasicOperation.tableScan("A"), BasicOperation
						.tableScan("B"), new JoinCondition("A", "age", "B",
						"old"));

		assertEquals("size 8", result.size(), 8);
		assertEquals("content", result.get(0).getKey("age"), "18");
		assertEquals("content", result.get(0).getKey("height"), "18");
		assertEquals("content", result.get(1).getKey("old"), "18");
		assertEquals("content", result.get(1).getKey("year"), "18");
		assertEquals("content", result.get(1).getKey("sex"), "male");

		assertEquals("content", result.get(2).getKey("age"), "16");
		assertEquals("content", result.get(2).getKey("height"), "172");
		assertEquals("content", result.get(3).getKey("old"), "16");
		assertEquals("content", result.get(3).getKey("year"), "16");
		assertEquals("content", result.get(3).getKey("sex"), "female");

		assertEquals("content", result.get(4).getKey("age"), "16");
		assertEquals("content", result.get(4).getKey("height"), "172");
		assertEquals("content", result.get(5).getKey("old"), "16");
		assertEquals("content", result.get(5).getKey("year"), "18");
		assertEquals("content", result.get(5).getKey("sex"), "male");
		
		assertEquals("content", result.get(6).getKey("age"), "18");
		assertEquals("content", result.get(6).getKey("height"), "170");
		assertEquals("content", result.get(7).getKey("old"), "18");
		assertEquals("content", result.get(7).getKey("year"), "18");
		assertEquals("content", result.get(7).getKey("sex"), "male");

		DataManager.exit();
	}

	@Test
	public void testIndex_join() {
		try {
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "170");
			record.put("height", "170");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "170");
			dataFile.insertRecord(record);

			dataFile.createIndex("__A.age", "age");

			fileName = "B";
			descriptor = new HashMap<String, Integer>();
			descriptor.put("sex", 50);
			descriptor.put("old", 50);
			descriptor.put("year", 50);
			dataFile = DataManager.createFile(fileName, descriptor);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "18");
			record.put("year", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "female");
			record.put("old", "16");
			record.put("year", "16");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "16");
			record.put("year", "18");
			dataFile.insertRecord(record);

			// tables
			String tableNames[] = new String[2];
			tableNames[0] = "A";
			tableNames[1] = "B";

			// records
			ArrayList<Record> records = BasicOperation.tableScan("B");
			ArrayList<Record> result = BasicOperation.index_join(records,
					new JoinCondition("A", "age", "B", "old"), tableNames);

			assertEquals("size 8", result.size(), 8);
			assertEquals("content", result.get(0).getKey("age"), "18");
			assertEquals("content", result.get(0).getKey("height"), "18");
			assertEquals("content", result.get(1).getKey("old"), "18");
			assertEquals("content", result.get(1).getKey("year"), "18");
			assertEquals("content", result.get(1).getKey("sex"), "male");

			assertEquals("content", result.get(2).getKey("age"), "18");
			assertEquals("content", result.get(2).getKey("height"), "170");
			assertEquals("content", result.get(3).getKey("old"), "18");
			assertEquals("content", result.get(3).getKey("year"), "18");
			assertEquals("content", result.get(3).getKey("sex"), "male");

			assertEquals("content", result.get(4).getKey("age"), "16");
			assertEquals("content", result.get(4).getKey("height"), "172");
			assertEquals("content", result.get(5).getKey("old"), "16");
			assertEquals("content", result.get(5).getKey("year"), "16");
			assertEquals("content", result.get(5).getKey("sex"), "female");

			assertEquals("content", result.get(6).getKey("age"), "16");
			assertEquals("content", result.get(6).getKey("height"), "172");
			assertEquals("content", result.get(7).getKey("old"), "16");
			assertEquals("content", result.get(7).getKey("year"), "18");
			assertEquals("content", result.get(7).getKey("sex"), "male");

			DataManager.exit();
		} catch (BasicOperationException e) {
			// TODO Auto-generated catch block
			fail("should not throw exception");
		}
	}

	@Test
	public void testIndex_join_both() {
		try {
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "170");
			record.put("height", "170");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "170");
			dataFile.insertRecord(record);

			dataFile.createIndex("__A.age", "age");

			fileName = "B";
			descriptor = new HashMap<String, Integer>();
			descriptor.put("sex", 50);
			descriptor.put("old", 50);
			descriptor.put("year", 50);
			dataFile = DataManager.createFile(fileName, descriptor);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "18");
			record.put("year", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "female");
			record.put("old", "16");
			record.put("year", "16");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "16");
			record.put("year", "18");
			dataFile.insertRecord(record);

			dataFile.createIndex("__B.old", "old");

			ArrayList<Record> result = BasicOperation
					.index_join_both(new JoinCondition("A", "age", "B", "old"));

			assertEquals("size 8", result.size(), 8);
			assertEquals("content", result.get(0).getKey("age"), "16");
			assertEquals("content", result.get(0).getKey("height"), "172");
			assertEquals("content", result.get(1).getKey("old"), "16");
			assertEquals("content", result.get(1).getKey("year"), "16");
			assertEquals("content", result.get(1).getKey("sex"), "female");

			assertEquals("content", result.get(2).getKey("age"), "16");
			assertEquals("content", result.get(2).getKey("height"), "172");
			assertEquals("content", result.get(3).getKey("old"), "16");
			assertEquals("content", result.get(3).getKey("year"), "18");
			assertEquals("content", result.get(3).getKey("sex"), "male");

			assertEquals("content", result.get(4).getKey("age"), "18");
			assertEquals("content", result.get(4).getKey("height"), "18");
			assertEquals("content", result.get(5).getKey("old"), "18");
			assertEquals("content", result.get(5).getKey("year"), "18");
			assertEquals("content", result.get(5).getKey("sex"), "male");

			assertEquals("content", result.get(6).getKey("age"), "18");
			assertEquals("content", result.get(6).getKey("height"), "170");
			assertEquals("content", result.get(7).getKey("old"), "18");
			assertEquals("content", result.get(7).getKey("year"), "18");
			assertEquals("content", result.get(7).getKey("sex"), "male");

			DataManager.exit();
		} catch (BasicOperationException e) {
			// TODO Auto-generated catch block
			fail("should not throw exception");
		}
	}

	@Test
	public void testFilter() {
		// one table
		// A.age = 18
		// A.height = 172
		{
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "170");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "170");
			dataFile.insertRecord(record);

			ArrayList<Record> result = BasicOperation.tableScan("A");
			ArrayList<JoinCondition> joinConditions = new ArrayList<JoinCondition>();
			ArrayList<NonJoinCondition> nonJoinConditions = new ArrayList<NonJoinCondition>();
			nonJoinConditions.add(new NonJoinCondition("A", "age", "18"));
			nonJoinConditions.add(new NonJoinCondition("A", "height", "172"));
			String tableNames[] = new String[1];
			tableNames[0] = "A";
			result = BasicOperation.filter(result, joinConditions,
					nonJoinConditions, tableNames);

			assertEquals("size 1", result.size(), 1);
			assertEquals("content", result.get(0).getKey("age"), "18");
			assertEquals("content", result.get(0).getKey("height"), "172");
			DataManager.exit();
		}
		// one table
		// A.age = 18
		{
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "170");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "170");
			dataFile.insertRecord(record);

			ArrayList<Record> result = BasicOperation.tableScan("A");
			ArrayList<JoinCondition> joinConditions = new ArrayList<JoinCondition>();
			ArrayList<NonJoinCondition> nonJoinConditions = new ArrayList<NonJoinCondition>();
			nonJoinConditions.add(new NonJoinCondition("A", "age", "18"));
			String tableNames[] = new String[1];
			tableNames[0] = "A";
			result = BasicOperation.filter(result, joinConditions,
					nonJoinConditions, tableNames);

			assertEquals("size 2", result.size(), 2);
			assertEquals("content", result.get(0).getKey("age"), "18");
			assertEquals("content", result.get(0).getKey("height"), "172");
			assertEquals("content", result.get(1).getKey("age"), "18");
			assertEquals("content", result.get(1).getKey("height"), "170");
			DataManager.exit();
		}

		// one table
		// A.age = A.height
		// A.age = 18
		{
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "170");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "170");
			dataFile.insertRecord(record);

			ArrayList<Record> result = BasicOperation.tableScan("A");
			ArrayList<JoinCondition> joinConditions = new ArrayList<JoinCondition>();
			joinConditions.add(new JoinCondition("A", "age", "A", "height"));
			ArrayList<NonJoinCondition> nonJoinConditions = new ArrayList<NonJoinCondition>();
			nonJoinConditions.add(new NonJoinCondition("A", "age", "18"));
			String tableNames[] = new String[1];
			tableNames[0] = "A";
			result = BasicOperation.filter(result, joinConditions,
					nonJoinConditions, tableNames);

			assertEquals("size 1", result.size(), 1);
			assertEquals("content", result.get(0).getKey("age"), "18");
			assertEquals("content", result.get(0).getKey("height"), "18");
			DataManager.exit();
		}

		// one table
		// A.age = A.height
		{
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "170");
			record.put("height", "170");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "170");
			dataFile.insertRecord(record);

			ArrayList<Record> result = BasicOperation.tableScan("A");
			ArrayList<JoinCondition> joinConditions = new ArrayList<JoinCondition>();
			joinConditions.add(new JoinCondition("A", "age", "A", "height"));
			ArrayList<NonJoinCondition> nonJoinConditions = new ArrayList<NonJoinCondition>();
			String tableNames[] = new String[1];
			tableNames[0] = "A";
			result = BasicOperation.filter(result, joinConditions,
					nonJoinConditions, tableNames);

			assertEquals("size 2", result.size(), 2);
			assertEquals("content", result.get(0).getKey("age"), "170");
			assertEquals("content", result.get(0).getKey("height"), "170");
			assertEquals("content", result.get(1).getKey("age"), "18");
			assertEquals("content", result.get(1).getKey("height"), "18");
			DataManager.exit();
		}

		// two table
		// A(age,height)
		// B(sex,old,year)
		// A.age = B.old
		// A.height = "18"
		// B.sex = "male"
		// A.age = A.height
		// B.old = B.year

		{
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "170");
			record.put("height", "170");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "170");
			dataFile.insertRecord(record);

			fileName = "B";
			descriptor = new HashMap<String, Integer>();
			descriptor.put("sex", 50);
			descriptor.put("old", 50);
			descriptor.put("year", 50);
			dataFile = DataManager.createFile(fileName, descriptor);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "18");
			record.put("year", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "female");
			record.put("old", "16");
			record.put("year", "16");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "16");
			record.put("year", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "18");
			record.put("year", "14");
			dataFile.insertRecord(record);

			// cartesian_product
			ArrayList<Record> records1 = BasicOperation.tableScan("A");
			ArrayList<Record> records2 = BasicOperation.tableScan("B");
			ArrayList<Record> result = BasicOperation.cartesian_product(
					records1, records2);

			// A.age = B.old
			// A.age = A.height
			// B.old = B.year

			// join condition
			ArrayList<JoinCondition> joinConditions = new ArrayList<JoinCondition>();
			joinConditions.add(new JoinCondition("A", "age", "A", "height"));
			joinConditions.add(new JoinCondition("B", "year", "B", "old"));
			joinConditions.add(new JoinCondition("A", "age", "B", "old"));

			// A.height = "18"
			// B.sex = "male"

			// non join condition
			ArrayList<NonJoinCondition> nonJoinConditions = new ArrayList<NonJoinCondition>();
			nonJoinConditions.add(new NonJoinCondition("A", "height", "18"));
			nonJoinConditions.add(new NonJoinCondition("B", "sex", "male"));

			// tables
			String tableNames[] = new String[2];
			tableNames[0] = "A";
			tableNames[1] = "B";

			// filter
			result = BasicOperation.filter(result, joinConditions,
					nonJoinConditions, tableNames);

			assertEquals("size 2", result.size(), 2);

			assertEquals("content", result.get(0).getKey("age"), "18");
			assertEquals("content", result.get(0).getKey("height"), "18");
			assertEquals("content", result.get(1).getKey("sex"), "male");
			assertEquals("content", result.get(1).getKey("old"), "18");
			assertEquals("content", result.get(1).getKey("year"), "18");

			DataManager.exit();
		}

		// two table
		// A(age,height)
		// B(sex,old,year)
		// A.age = B.old
		// B.sex = "male"
		// A.age = A.height
		// B.old = B.year

		{
			String fileName = "A";
			Map<String, Integer> descriptor = new HashMap<String, Integer>();
			descriptor.put("age", 50);
			descriptor.put("height", 50);
			DataFile dataFile = DataManager.createFile(fileName, descriptor);
			Map<String, String> record = new HashMap<String, String>();
			record.put("age", "170");
			record.put("height", "170");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "16");
			record.put("height", "172");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("age", "18");
			record.put("height", "170");
			dataFile.insertRecord(record);

			fileName = "B";
			descriptor = new HashMap<String, Integer>();
			descriptor.put("sex", 50);
			descriptor.put("old", 50);
			descriptor.put("year", 50);
			dataFile = DataManager.createFile(fileName, descriptor);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "18");
			record.put("year", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "female");
			record.put("old", "16");
			record.put("year", "16");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "16");
			record.put("year", "18");
			dataFile.insertRecord(record);
			record = new HashMap<String, String>();
			record.put("sex", "male");
			record.put("old", "18");
			record.put("year", "14");
			dataFile.insertRecord(record);

			// cartesian_product
			ArrayList<Record> records1 = BasicOperation.tableScan("A");
			ArrayList<Record> records2 = BasicOperation.tableScan("B");
			ArrayList<Record> result = BasicOperation.cartesian_product(
					records1, records2);

			// A.age = B.old
			// A.age = A.height
			// B.old = B.year

			// join condition
			ArrayList<JoinCondition> joinConditions = new ArrayList<JoinCondition>();
			joinConditions.add(new JoinCondition("A", "age", "A", "height"));
			joinConditions.add(new JoinCondition("B", "year", "B", "old"));
			joinConditions.add(new JoinCondition("A", "age", "B", "old"));

			// A.height = "18"
			// B.sex = "male"

			// non join condition
			ArrayList<NonJoinCondition> nonJoinConditions = new ArrayList<NonJoinCondition>();
			nonJoinConditions.add(new NonJoinCondition("B", "sex", "male"));

			// tables
			String tableNames[] = new String[2];
			tableNames[0] = "A";
			tableNames[1] = "B";

			// filter
			result = BasicOperation.filter(result, joinConditions,
					nonJoinConditions, tableNames);

			assertEquals("size 2", result.size(), 2);

			assertEquals("content", result.get(0).getKey("age"), "18");
			assertEquals("content", result.get(0).getKey("height"), "18");
			assertEquals("content", result.get(1).getKey("sex"), "male");
			assertEquals("content", result.get(1).getKey("old"), "18");
			assertEquals("content", result.get(1).getKey("year"), "18");

			DataManager.exit();
		}

	}

}
